package com.xzzz.expand.es.core;

import cn.hutool.core.util.StrUtil;
import com.xzzz.expand.es.core.common.DiscoverEnum;
import com.xzzz.expand.es.core.common.EsException;
import com.xzzz.expand.es.core.EsProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.elasticsearch.client.*;
import org.elasticsearch.client.core.MainResponse;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;


/**
 * ElasticSearch 配置类
 *
 * @author xzzz
 *
 */
@Slf4j
@Configuration
public class EsClientConfig {

    /**
     * 注入 ES 配置对象 {@link EsProperties}
     */
    @Autowired
    private EsProperties esProperties;

    /**
     * 构造 RestClientBuilder
     *
     * @return RestClientBuilder
     */
    @Bean
    @SuppressWarnings("all")
    public RestClientBuilder restClientBuilder() {
        RestClientBuilder restClientBuilder;
        // 单机模式, 直接取主机地址
        if (DiscoverEnum.SINGLE.getType().equals(esProperties.getDiscoveryType()) && esProperties.getSingle() != null) {
            restClientBuilder = RestClient
                    .builder(buildHttpHost(esProperties.getSingle().getNode()));
        }
        // 集群模式，构造多个节点
        else if (DiscoverEnum.CLUSTER.getType().equals(esProperties.getDiscoveryType()) && esProperties.getCluster() != null) {
            restClientBuilder = RestClient
                    .builder(buildHttpHost(esProperties.getCluster().getNodes()));
        } else {
            throw new IllegalArgumentException("错误的配置类型:" + esProperties.getDiscoveryType());
        }

        setFailureListener(restClientBuilder);
        setRequestConfigCallback(restClientBuilder);

        restClientBuilder.setHttpClientConfigCallback(configCallback -> {
            /**
             * 设置用户名密码
             */
            if (StrUtil.isNotBlank(esProperties.getUsername()) && StrUtil.isNotBlank(esProperties.getPassword())) {
                CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(esProperties.getUsername(), esProperties.getPassword()));
                configCallback.setDefaultCredentialsProvider(credentialsProvider);
            } else {
                //禁用身份验证缓存
                configCallback.disableAuthCaching();
            }

            /**
             * 显式设置 keepAliveStrategy, 返回连接允许的空闲时间
             */
            configCallback.setKeepAliveStrategy((httpResponse, httpContext) -> TimeUnit.MINUTES.toMillis(10));
            /**
             * 显式开启 tcp keepalive, 默认是 {@link IOReactorConfig#DEFAULT}
             * 默认配置项
             * this.selectInterval = 1000;
             * this.shutdownGracePeriod = 500;
             * this.interestOpQueued = false;
             * this.ioThreadCount = getDefaultMaxIoThreadCount();
             * this.soTimeout = 0;
             * this.soReuseAddress = false;
             * this.soLinger = -1;
             * this.soKeepAlive = false;
             * this.tcpNoDelay = true;
             * this.connectTimeout = 0;
             * this.sndBufSize = 0;
             * this.rcvBufSize = 0;
             * this.backlogSize = 0;
             */
            configCallback.setDefaultIOReactorConfig(IOReactorConfig.custom().setSoKeepAlive(true).build());
            return configCallback;
        });

        return restClientBuilder;
    }

    /**
     * 创建 RestHighLevelClient 对象
     *
     * @param restClientBuilder restClientBuilder
     * @return RestHighLevelClient
     */
    @Bean
    public RestHighLevelClient restHighLevelClient(@Autowired RestClientBuilder restClientBuilder) {
        RestHighLevelClient restHighLevelClient = new RestHighLevelClient(restClientBuilder);
        try {
            MainResponse response = restHighLevelClient.info(RequestOptions.DEFAULT);
            log.info("[ELSEARCH] 初始化 HighClient 成功, 集群UUID:{}, 节点名称:{}, 版本:{}",
                    response.getClusterUuid(),
                    response.getNodeName(),
                    response.getVersion().getNumber());
            return restHighLevelClient;
        } catch (IOException e) {
            e.printStackTrace();
            throw new EsException("初始化 RestHighLevelClient 失败");
        }
    }


    /*
     * 设置一个监听器，每次节点出现故障时都会收到通知，以防需要采取措施，
     */
    private void setFailureListener(RestClientBuilder restClientBuilder) {
        restClientBuilder.setFailureListener(new RestClient.FailureListener() {
            @Override
            public void onFailure(Node node) {
                log.error("[ELSEARCH] 连接ES节点失败: {}", node.toString());
            }
        });
    }


    /*
     * 设置允许修改默认请求配置的回调
     */
    private void setRequestConfigCallback(RestClientBuilder restClientBuilder) {
        restClientBuilder.setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder
                // 设置从connect Manager(连接池)获取Connection 超时时间，单位毫秒。HttpClient中的要用连接时尝试从连接池中获取，
                // 若是在等待了一定的时间后还没有获取到可用连接（比如连接池中没有空闲连接了）则会抛出获取连接超时异常。
                .setConnectionRequestTimeout(esProperties.getConnectRequestTimeout())
                // 请求获取数据的超时时间(即响应时间)，单位毫秒。即在与目标url建立了连接之后，等待返回数据的时间的最大时间， 如果超出
                // 时间没有响应，就会抛出socketTimeoutException异常。
                .setSocketTimeout(esProperties.getSocketTimeout())
                // 设置连接超时时间，单位毫秒。指的是连接目标url的连接超时时间，即客服端发送请求到与目标url建立起连接的最大时间。
                // 如果在该时间范围内还没有建立起连接，会抛出connectionTimeOut异常。
                .setConnectTimeout(esProperties.getConnectTimeout())
        );
    }


    /**
     * 创建 HttpHost
     *
     * @param str IP端口字符串
     */
    private HttpHost buildHttpHost(String str) {
        String[] ipAndPort;
        if (StrUtil.isBlank(str) || (ipAndPort = str.split(":")).length != 2) {
            log.error("[ELSEARCH] ES地址配置错误: {}", str);
            throw new IllegalArgumentException("ES地址配置错误: " + str);
        }
        return new HttpHost(ipAndPort[0], Integer.parseInt(ipAndPort[1]), esProperties.getScheme());
    }

    /**
     * HttpHost 数组
     *
     * @param strs IP端口字符串数组
     */
    private HttpHost[] buildHttpHost(String[] strs) {
        if (strs.length == 0) {
            log.error("[ELSEARCH] ES地址配置错误: {}", Arrays.toString(strs));
            throw new IllegalArgumentException("ES地址配置错误: " + Arrays.toString(strs));
        }

        HttpHost[] hosts = new HttpHost[strs.length];
        for (int i = 0; i < strs.length; i++) {
            hosts[i] = buildHttpHost(strs[i]);
        }
        return hosts;
    }

}
